import sys

from PyQt5.QtCore import QThread, pyqtSignal, QRegExp
from PyQt5.QtGui import QTextCursor, QRegExpValidator
from PyQt5 import QtWidgets
from PyQt5.QtWidgets import QMessageBox

from kafka import KafkaConsumer
from kafka import errors

from utils.utils import *


class ConsumerGui(QtWidgets.QWidget):
    """KAFKA消费者"""

    def __init__(self):
        super().__init__()

        self.consumer_thread = None
        self.setWindowTitle("KAFKA消费者模拟工具")

        self.host = QtWidgets.QLineEdit(self)
        self.host.setPlaceholderText("KAFKA服务器地址")
        self.host.setInputMask("000.000.000.000;_")

        self.port = QtWidgets.QLineEdit(self)
        self.port.setPlaceholderText("19092")
        self.port.setValidator(QLinedit_format(is_letter=False))  # 进允许输入数字

        self.topic = QtWidgets.QLineEdit(self)
        self.topic.setPlaceholderText("TOPIC")

        self.connect_button = QtWidgets.QPushButton("连接", self)
        self.close_button = QtWidgets.QPushButton("断开", self)
        self.close_button.setEnabled(False)
        self.clear_button = QtWidgets.QPushButton("清空消息", self)

        self.receive_msg_edit = QtWidgets.QTextEdit(self)
        self.receive_msg_edit.setReadOnly(True)

        layout1 = QtWidgets.QHBoxLayout()
        layout1.addWidget(QtWidgets.QLabel("地址："))
        layout1.addWidget(self.host)
        layout1.addWidget(QtWidgets.QLabel("端口："))
        layout1.addWidget(self.port)
        layout1.addWidget(QtWidgets.QLabel("TOPIC："))
        layout1.addWidget(self.topic)
        layout1.addWidget(self.connect_button)
        layout1.addWidget(self.close_button)
        layout1.addWidget(self.clear_button)

        layout2 = QtWidgets.QVBoxLayout()
        layout2.addWidget(QtWidgets.QLabel("接收到的消息："))
        layout2.addWidget(self.receive_msg_edit)

        layout3 = QtWidgets.QVBoxLayout()
        layout3.addLayout(layout1)
        layout3.addLayout(layout2)
        self.setLayout(layout3)

        self.ui_init()

        self.connect_button.clicked.connect(self.connect_to_kafka)
        self.clear_button.clicked.connect(self.clear_received_messages)

    def connect_to_kafka(self):
        """
        连接到kafka
        @return:
        """
        host = self.host.text()
        port = self.port.text()
        topic = self.topic.text()
        if not host or not port or not topic:
            QMessageBox.critical(self, "错误", "请填写完整的信息")
            return
        try:
            self.connect_button.setEnabled(False)
            self.close_button.setEnabled(True)
            self.consumer_thread = KafkaConsumerThread(
                f"{self.host.text()}:{self.port.text()}", self.topic.text()
            )
            self.consumer_thread.message_received.connect(self.handle_message_received)
            self.consumer_thread.start()
        except errors.NoBrokersAvailable:
            self.connect_button.setEnabled(True)
            self.close_button.setEnabled(False)
            self.show_message()

    def closeEvent(self, event):
        print("close")
        if self.consumer_thread:
            self.consumer_thread.terminate()
        event.accept()

    # def __del__(self):  # 重写析构函数
    #     if self.consumer_thread:
    #         self.consumer_thread.terminate()

    def show_message(self):
        QMessageBox.critical(self, "错误", "系统错误")

    def handle_message_received(self, message):
        self.receive_msg_edit.insertPlainText(message + "\n\n")
        self.receive_msg_edit.moveCursor(QTextCursor.End)
        self.receive_msg_edit.ensureCursorVisible()

    def clear_received_messages(self):
        self.receive_msg_edit.clear()

    def ui_init(self):
        self.host.setText("192.168.120.221")
        self.port.setText("9094")
        self.topic.setText("SYSTEM_MASTER_IN")


class KafkaConsumerThread(QThread):
    message_received = pyqtSignal(str)

    def __init__(self, kafka_url, topic):
        super().__init__()
        self.kafka_url = kafka_url
        self.topic = topic
        self.consumer = None

    def __exit__(self, exc_type, exc_val, exc_tb):
        """

        @param exc_type:
        @param exc_val:
        @param exc_tb:
        @return:
        """
        if self.consumer:
            self.consumer.close()

    def __del__(self):  # 重写析构函数
        print("del")
        if self.consumer:
            self.consumer.close()

    def closeEvent(self, event):
        if self.consumer:
            self.consumer.close()
        event.accept()

    def run(self):
        try:
            self.consumer = KafkaConsumer(
                self.topic,
                bootstrap_servers=self.kafka_url,
                auto_offset_reset="earliest",
                enable_auto_commit=True,
            )
            for message in self.consumer:
                self.message_received.emit(message.value.decode("utf-8"))
        except Exception as e:
            self.message_received.emit(f"ERROR: {str(e)}")

    def terminate(self):
        if self.consumer:
            self.consumer.close()
        super().terminate()


if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    window = ConsumerGui()
    window.show()
    sys.exit(app.exec_())
